package m3;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.Scanner;

public class Producer {
    public static void main(String[] args) throws MQClientException {
        //创建事务消息生产者
        TransactionMQProducer p = new TransactionMQProducer("prod-group3");
        //设置 name server
        p.setNamesrvAddr("192.168.64.141:9876");

        //启动
        p.start();
        //2.处理rocketmq的事务回查
        p.setTransactionListener(new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                //模拟网络不稳定,不能向服务器提交或者回滚消息
                if (Math.random()<0.7){
                    System.out.println("网络中断,无法向服务器提交或回滚: ");
                    return LocalTransactionState.UNKNOW; //未知,当前方法中,一般不会出现
                }

                System.out.println("执行本地业务,参数: "+arg);
                if(Math.random()<0.5){
                    System.out.println("本地事务执行成功: ");
                    return LocalTransactionState.COMMIT_MESSAGE; //提交消息,通知服务器消息可以投递
                }else {
                    System.out.println("本地事务执行失败: ");
                    return LocalTransactionState.ROLLBACK_MESSAGE; //回滚消息
                }


                //return LocalTransactionState.UNKNOW; //未知,当前方法中,一般不会出现
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                System.out.println("rocketmq 服务器正在回查事务状态");
                /**
                 * 上面本地事务还没有执行结束
                 * 无法获取事务状态
                 */
                if (Math.random()<0.7){
                    System.out.println("网络中断,无法向服务器提交或回滚: ");
                    return LocalTransactionState.UNKNOW; //未知,当前方法中,一般不会出现
                }
                if(Math.random()<0.5){
                    System.out.println("本地事务执行成功: ");
                    return LocalTransactionState.COMMIT_MESSAGE; //提交消息,通知服务器消息可以投递
                }else {
                    System.out.println("本地事务执行失败: ");
                    return LocalTransactionState.ROLLBACK_MESSAGE; //回滚消息
                }
            }
        });

        // 设置事务消息监听器
        while(true){
            System.out.println("输入消息: ");
            String s = new Scanner(System.in).nextLine();
            Message msg = new Message("Topic3", s.getBytes());
            TransactionSendResult r = p.sendMessageInTransaction(msg, "业务数据参数");
            System.out.println("事务消息结果是: "+r);
        }
            //1.执行本地事务


        //发送事务消息,触发监听器



    }
}
